package test.threads.queue.unstressed;

import java.util.concurrent.Executor;

import com.lmax.disruptor.EventFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import test.helpers.Config;
import test.threads.queue.common.BaseEnqueuer;
import test.threads.queue.common.IUsefulWorkEmulator;
import test.threads.queue.stressed.AbstractTask;
import test.threads.queue.stressed.cpu.LongValueMessage;

/**
 * new version of {@link QueueBenchmarkingTask}
 *
 * @author cheremin
 * @since 24.03.12,  14:52
 */
public class QueueTask extends AbstractTask<LongValueMessage> {
    private static final Log log = LogFactory.getLog( QueueTask.class );

    private final IBoundedQueueFactory<LongValueMessage, ? extends IQueue<LongValueMessage>> queueFactory;
    private final IQueue<LongValueMessage> queue;

    public QueueTask( final Config config ) throws Exception {
        super( config );

        final Class<IBoundedQueueFactory> clazz = config.getAsClass( "task.queue-factory" );

        this.queueFactory = ( IBoundedQueueFactory ) clazz.getField( "FACTORY" ).get( null );

        this.queue = queueFactory.create( bufferSize() );
    }


    @Override
    protected BaseEnqueuer _initialize( final Executor threadPool,
                                        final IUsefulWorkEmulator<LongValueMessage>[] emulators,
                                        final EventFactory<LongValueMessage> factory ) throws Exception {
        return new Enqueuer();
    }

    @Override
    protected long dequeueBatchEvents( final int eventsInBatch ) throws Exception {
        long sum = 0;
        for ( int i = 0; i < eventsInBatch; i++ ) {
            final LongValueMessage msg = queue.dequeue();
            sum += msg.value;
        }
        return sum;
    }

    @Override
    public int workersRequired() {
        return 1;
    }

    @Override
    public String toString() {
        return String.format(
                "%s[%s[%d]]",
                getClass().getSimpleName(),
                queueFactory,
                bufferSize()
        );
    }

    private class Enqueuer extends BaseEnqueuer<LongValueMessage> {
        private Enqueuer() {
            super( eventsInBatch() );
        }

        @Override
        protected void enqueueEventsBatch( final int eventsInBatch ) throws Exception {
            final EventFactory<LongValueMessage> factory = eventFactory();
            for ( int i = 0; i < eventsInBatch; i++ ) {
                final LongValueMessage event = factory.newInstance();
                event.value = i;
                queue.enqueue( event );
            }
        }
    }
}
